use chrono::{DateTime, Datelike, Local, NaiveDate, NaiveDateTime, TimeDelta, Timelike, Utc};
use polars::datatypes::{AnyValue, DataType, TimeUnit};
use polars::frame::DataFrame;
use polars::prelude::{Column, NamedFrom, ParquetWriter, PlSmallStr, Series};
use polars::sql::SQLContext;
use polars_lazy::frame::IntoLazy;
use std::collections::HashMap;
use std::fs::File;
use std::path::PathBuf;
use std::sync::{Arc, LazyLock, RwLock};
use visit_dd_core::field::{
    VddField, VddFieldRef, VddFieldType, C_FT_BOOL, C_FT_DATE, C_FT_DATETIME, C_FT_FLOAT, FT_INT,
    C_FT_TIME, FT_STRING,
};
use visit_dd_core::id_factory::ID_FACTORY;
use visit_dd_core::model::{VddModel, VddSchema};
use visit_dd_core::proj::APP_CONFIG;
use visit_dd_core::query::{SqlQueryResult, WebQueryParams};
use visit_dd_core::record::{VddDisplay, VddRecord};
use visit_dd_core::value::VddValue;
use visit_dd_core::visit::VisitGroup;
use visit_dd_core::{i18n, Astr, FieldId, UserId, T_FORMAT};
use visit_dd_db::sqlx::SqlxDataStoreTrait;
use visit_dd_db::{cache, record_db, subscribe};

/// 数据集缓存
pub(crate) struct ModelDataFrame {
    pub model_uid: UserId,
    pub model_tag: Astr,
    pub df: DataFrame,
}

static DATA_FRAMES: LazyLock<RwLock<Vec<Arc<ModelDataFrame>>>> =
    LazyLock::new(|| RwLock::new(Vec::<Arc<ModelDataFrame>>::new()));

pub(crate) fn export_schema(schema: &VddSchema) -> Result<u64, Astr> {
    let path = parquet_cache_path(schema.model.id);
    if path.exists() && path.is_file() {
        std::fs::remove_file(&path).expect(i18n!(1021).as_ref());
    }
    let file = File::create(&path).unwrap();
    let cols = schema
        .fields
        .iter()
        .map(|x| {
            let tag = schema.tags.get(&x.id.0).unwrap();
            let series = Column::new_empty(PlSmallStr::from_str(tag), &field_to_pdt(x));
            Column::from(series)
        })
        .collect::<Vec<_>>();
    let mut df = DataFrame::new(cols).unwrap();
    ParquetWriter::new(file)
        .finish(&mut df)
        .map_err(|e| e.to_string().into())
}

/// 获取一个模型的parquet缓存文件
pub(crate) fn parquet_cache_path(model_id: u64) -> PathBuf {
    PathBuf::from(format!("{}/{}.parquet", APP_CONFIG.db_dir, model_id))
    // info!("parquet_cache_path {:?}", path);
}

/// 刷新一个模型的parquet缓存
pub(crate) async fn refresh_dataframe_cache(model: &VddModel) {
    let schema = cache::get_schema(model.id).await.unwrap();
    let need_refresh = subscribe::get_parquet_refresh_flag(model).await;
    if need_refresh {
        let records = match model.store.clone() {
            Some(ds) => {
                let wqp = WebQueryParams {
                    model_id: Some(model.id),
                    page_index: 1,
                    page_size: 10000,
                    last_id: 0,
                    new_to_old: false,
                    filters: Vec::new(),
                };
                let vdds = ds.sqlx_get_page(&wqp, &schema).await;
                vdds.data.iter().map(|vdd| vdd.record()).collect::<Vec<_>>()
            }
            None => record_db::select_records_all(model.id).await,
        };
        let r = export_with_record(&schema, &records).await;
        if r.is_err() {
            tracing::error!("export model and record error:{}", r.unwrap_err());
        }
    }
}

/// 在本地存储的数据模型中执行sql语句 将本地存储的记录转成parquet文件，加载到内存中，用polars查询，再将结果转成record
pub(crate) async fn try_proxy_sql(uid: &UserId, sql: &str) -> Result<SqlQueryResult, Astr> {
    tracing::info!("sql {}", sql);
    let mut sc = SQLContext::new();
    let data_frames = LazyLock::force(&DATA_FRAMES).read().unwrap();
    data_frames.iter().for_each(|x| {
        if x.model_uid == uid.to_owned() {
            sc.register(&x.model_tag, x.df.to_owned().lazy());
            tracing::info!("{}", i18n!(1023, &x.model_tag));
        }
    });
    let sql_df = sc.execute(sql);
    if sql_df.is_err() {
        return Err(Arc::from(sql_df.err().unwrap().to_string()));
    }
    let sql_df = sql_df.unwrap().collect().unwrap();
    df_to_records(&sql_df)
}

/// 在本地存储的数据模型中执行sql语句 将本地存储的记录转成parquet文件，加载到内存中，用polars查询，再将结果转成record
pub(crate) async fn try_run_share_sql(
    auths: &[Arc<VisitGroup>],
    tags: &HashMap<u64, Astr>,
    sql: &str,
) -> Result<SqlQueryResult, Astr> {
    let mut sc = SQLContext::new();
    let data_frames = LazyLock::force(&DATA_FRAMES).read().unwrap();
    auths
        .iter()
        .map(|g| tags.get(&g.model_id))
        .filter(|tag| tag.is_some())
        .map(|tag| tag.unwrap())
        .for_each(|tag| {
            let mdf = data_frames.iter().find(|mdf| &mdf.model_tag == tag);
            if mdf.is_some() {
                sc.register(&tag, mdf.unwrap().df.to_owned().lazy());
                tracing::info!("{}", i18n!(1023, &tag));
            }
        });
    let sql_df = sc.execute(sql);
    if sql_df.is_err() {
        return Err(Arc::from(sql_df.err().unwrap().to_string()));
    }
    let sql_df = sql_df.unwrap().collect().unwrap();
    df_to_records(&sql_df)
}

fn field_to_pdt(field: &VddField) -> DataType {
    match field.field_type.as_ref() {
        VddFieldType::FtInt { .. } | VddFieldType::FtModel { .. } | VddFieldType::FtUserId => {
            if field.multi {
                DataType::List(DataType::Int64.into())
            } else {
                DataType::Int64
            }
        }
        VddFieldType::FtFloat { .. } => {
            if field.multi {
                DataType::List(DataType::Float64.into())
            } else {
                DataType::Float64
            }
        }
        VddFieldType::FtBool => {
            if field.multi {
                DataType::List(DataType::Boolean.into())
            } else {
                DataType::Boolean
            }
        }
        VddFieldType::FtDateTime(..) => {
            if field.multi {
                DataType::List(DataType::Datetime(TimeUnit::Microseconds, None).into())
            } else {
                DataType::Datetime(TimeUnit::Microseconds, None)
            }
        }
        VddFieldType::FtDate(..) => {
            if field.multi {
                DataType::List(DataType::Date.into())
            } else {
                DataType::Date
            }
        }
        VddFieldType::FtTime(..) => {
            if field.multi {
                DataType::List(DataType::Time.into())
            } else {
                DataType::Time
            }
        }
        _ => {
            if field.multi {
                DataType::List(DataType::String.into())
            } else {
                DataType::String
            }
        }
    }
}

fn field_type_from_datatype(datatype: &DataType) -> VddFieldType {
    match datatype {
        DataType::Boolean => VddFieldType::test_u16(C_FT_BOOL, None).unwrap(),
        DataType::Int64
        | DataType::Int32
        | DataType::Int16
        | DataType::Int8
        | DataType::UInt8
        | DataType::UInt64
        | DataType::UInt32
        | DataType::UInt16 => VddFieldType::test_u16(FT_INT, None).unwrap(),
        DataType::Float64 | DataType::Float32 => VddFieldType::test_u16(C_FT_FLOAT, None).unwrap(),
        DataType::Date => VddFieldType::test_u16(C_FT_DATE, None).unwrap(),
        DataType::Time => VddFieldType::test_u16(C_FT_TIME, None).unwrap(),
        DataType::Datetime(..) => VddFieldType::test_u16(C_FT_DATETIME, None).unwrap(),
        DataType::List(item_type) => field_type_from_datatype(item_type),
        DataType::String => VddFieldType::test_u16(FT_STRING, None).unwrap(),
        DataType::Binary => {
            panic!("Not supported")
        }
        DataType::BinaryOffset => {
            panic!("Not supported")
        }
        DataType::Duration(_) => {
            panic!("Not supported")
        }
        DataType::Null => VddFieldType::FtString {
            min_len: None,
            max_len: None,
            pattern: None,
        },
        DataType::Struct(_) => {
            panic!("Not supported")
        }
        DataType::Unknown(_) => {
            panic!("Not supported")
        }
        DataType::Int128 => {
            panic!("Not supported")
        }
        DataType::Decimal(_, _) => {
            panic!("Not supported")
        }
    }
}

fn field_from_pdt(dt: &DataType, model_id: u64, display_order: i8, label: &str) -> Arc<VddField> {
    let field_type = field_type_from_datatype(dt);
    let multi = match dt {
        DataType::List(..) => true,
        _ => false,
    };
    VddField {
        id: FieldId(ID_FACTORY.lock().unwrap().lazy_generate()),
        model_id,
        display_order,
        label: label.to_string(),
        field_type: field_type.into(),
        required: false,
        multi,
        unique: false,
        update_ts: Local::now().naive_local(),
        input_hint: None,
    }
    .into()
}

async fn records_to_df(schema: &VddSchema, records: &[Arc<VddRecord>]) -> DataFrame {
    let mut export_values = Vec::<Vec<AnyValue>>::new(); // 全部的数据
    for _ in 0..schema.fields.len() {
        // 预分配内存空间
        export_values.push(Vec::<AnyValue>::with_capacity(records.len()));
    }
    records.iter().for_each(|record| {
        // 批量转换值
        let mut col_idx = 0;
        schema.fields.iter().for_each(|field| {
            let fv = record.value.get(&field.id);
            if let Some(fv) = fv {
                export_values
                    .get_mut(col_idx)
                    .unwrap()
                    .push(fv_to_anyvalue(fv));
            } else {
                export_values.get_mut(col_idx).unwrap().push(AnyValue::Null);
            }
            col_idx += 1;
        });
    });
    let mut series = Vec::<Series>::new();
    let mut col_idx = 0;
    schema.fields.iter().for_each(|x| {
        let tag = schema.tags.get(&x.id.0).unwrap();
        let tag = PlSmallStr::from_str(tag);
        series.push(Series::new(tag, &export_values.get(col_idx).unwrap()));
        col_idx += 1;
    });
    let cols = series
        .iter()
        .map(|x| Column::from(x.clone()))
        .collect::<Vec<Column>>();
    let df = DataFrame::new(cols).unwrap();
    LazyLock::force(&DATA_FRAMES)
        .write()
        .unwrap()
        .push(Arc::new(ModelDataFrame {
            model_uid: schema.model.uid.clone(),
            model_tag: schema.get_model_tag().to_owned(),
            df: df.clone(),
        }));
    df
}

/// 查询出来的dataframe转换
pub(crate) fn df_to_records(df: &DataFrame) -> Result<SqlQueryResult, Astr> {
    // tracing::info!("{:?}", df);
    let mut records = Vec::<Arc<VddDisplay>>::new();
    let mut fields = Vec::<VddFieldRef>::new();
    let mut col_index: usize = 0;
    df.schema().iter().for_each(|item| {
        fields.push(Arc::new(VddField {
            id: FieldId(col_index as u64),
            model_id: 0,
            display_order: (col_index + 1) as i8,
            label: item.0.to_string(),
            field_type: Arc::new(field_type_from_datatype(item.1)),
            required: false,
            multi: false,
            unique: false,
            input_hint: None,
            update_ts: Local::now().naive_local(),
        }));
        col_index += 1;
    });
    let mut row_index: usize = 0;
    while row_index < df.shape().0 {
        col_index = 0;
        let mut record = VddRecord::new_default(UserId(0), 0);
        df.schema().iter().for_each(|item| {
            let col = df.column(&item.0).unwrap();
            record.value.insert(
                FieldId(col_index as u64),
                Arc::new(fv_from_anyvalue(&col.get(row_index).unwrap())),
            );
            col_index += 1;
        });
        let recordd = VddDisplay::from(Arc::new(record));
        records.push(Arc::new(recordd));
        row_index += 1;
    }
    Ok(SqlQueryResult {
        records,
        fields,
        title: Local::now().time().format(T_FORMAT).to_string().into(),
    })
}

/// 将模型和记录导出为parquet文件
pub(crate) async fn export_with_record(
    schema: &VddSchema,
    records: &[Arc<VddRecord>],
) -> Result<u64, Astr> {
    let path = parquet_cache_path(schema.model.id);
    if path.exists() && path.is_file() {
        std::fs::remove_file(&path).expect(i18n!(1021).as_ref())
    }
    let file = File::create(&path).unwrap();
    let mut df = records_to_df(schema, records).await;
    ParquetWriter::new(file)
        .finish(&mut df)
        .map_err(|e| Arc::from(e.to_string()))
}

/// 将上传parquet文件的解析结果存放进缓存
pub(crate) fn df_cache_put(file_key: u64, df: &DataFrame) {
    let mut data_frames = LazyLock::force(&DATA_FRAMES).write().unwrap();
    data_frames.push(Arc::new(ModelDataFrame {
        model_uid: UserId(file_key),
        model_tag: "".into(),
        df: df.clone(),
    }));
}

/// 从缓存中读出对应文件的解析结果，以便后续生成字段和记录
pub(crate) fn df_cache_get(file_key: u64) -> Option<Arc<ModelDataFrame>> {
    let mut r: Option<Arc<ModelDataFrame>> = None;
    let data_frames = LazyLock::force(&DATA_FRAMES).read().unwrap();
    data_frames.iter().for_each(|data_frame| {
        if data_frame.model_uid.0 == file_key {
            r = Some(data_frame.clone());
        }
    });
    r
}

/// 从缓存中删除不用的parquet文件解析结果
pub(crate) fn df_cache_remove(file_key: u64) {
    LazyLock::force(&DATA_FRAMES)
        .write()
        .unwrap()
        .retain(|data_frame| data_frame.model_uid.0 != file_key);
}

/// 从dataframe中生成字段列表
pub(crate) fn fields_from_df(model_id: u64, df: &DataFrame) -> Result<Vec<VddFieldRef>, Astr> {
    let mut result: Vec<VddFieldRef> = Vec::new();
    for x in df.fields() {
        let vdd_f = field_from_pdt(x.dtype(), model_id, (result.len() + 1) as i8, x.name());
        result.push(vdd_f);
    }
    Ok(result)
}

fn fv_from_anyvalue(av: &AnyValue) -> VddValue {
    match av {
        AnyValue::Null => VddValue::NULL,
        AnyValue::Boolean(b) => VddValue::S211(b.clone()),
        AnyValue::String(s) => VddValue::S101(s.to_string()),
        AnyValue::UInt8(i) => VddValue::S201(i.clone() as i64),
        AnyValue::UInt16(i) => VddValue::S201(i.clone() as i64),
        AnyValue::UInt32(i) => VddValue::S201(i.clone() as i64),
        AnyValue::UInt64(i) => VddValue::S201(i.clone() as i64),
        AnyValue::Int8(i) => VddValue::S201(i.clone() as i64),
        AnyValue::Int16(i) => VddValue::S201(i.clone() as i64),
        AnyValue::Int32(i) => VddValue::S201(i.clone() as i64),
        AnyValue::Int64(i) => VddValue::S201(i.clone()),
        AnyValue::Float32(f) => VddValue::S202(f.clone() as f64),
        AnyValue::Float64(f) => VddValue::S202(f.clone()),
        AnyValue::Date(d) => VddValue::S301(
            NaiveDate::from_ymd_opt(1970, 1, 1).unwrap() + TimeDelta::days(d.clone() as i64),
        ),
        AnyValue::Datetime(dt, _, _) => {
            VddValue::S303(Local::now().naive_local() + TimeDelta::nanoseconds(dt.clone()))
        }
        AnyValue::List(s) => {
            // 探测子元素是否是字符类型
            let mut value = s
                .str()
                .map(|ary| {
                    let mut vec = Vec::<String>::new();
                    for v in ary.into_iter() {
                        match v {
                            Some(vv) => {
                                vec.push(vv.to_string());
                            }
                            None => {}
                        }
                    }
                    VddValue::M101(vec)
                })
                .unwrap_or(VddValue::NULL);
            // 探测子元素是否是i8
            if value == VddValue::NULL {
                value = s
                    .i8()
                    .map(|ary| {
                        let mut vec = Vec::<i64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv as i64);
                                }
                                None => {}
                            }
                        }
                        VddValue::M201(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            // 探测子元素是否是i16
            if value == VddValue::NULL {
                value = s
                    .i16()
                    .map(|ary| {
                        let mut vec = Vec::<i64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv as i64);
                                }
                                None => {}
                            }
                        }
                        VddValue::M201(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            //
            if value == VddValue::NULL {
                value = s
                    .i32()
                    .map(|ary| {
                        let mut vec = Vec::<i64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv as i64);
                                }
                                None => {}
                            }
                        }
                        VddValue::M201(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            //
            if value == VddValue::NULL {
                value = s
                    .i64()
                    .map(|ary| {
                        let mut vec = Vec::<i64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv);
                                }
                                None => {}
                            }
                        }
                        VddValue::M201(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            //
            if value == VddValue::NULL {
                value = s
                    .u8()
                    .map(|ary| {
                        let mut vec = Vec::<i64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv as i64);
                                }
                                None => {}
                            }
                        }
                        VddValue::M201(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            //
            if value == VddValue::NULL {
                value = s
                    .u16()
                    .map(|ary| {
                        let mut vec = Vec::<i64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv as i64);
                                }
                                None => {}
                            }
                        }
                        VddValue::M201(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            //
            if value == VddValue::NULL {
                value = s
                    .u32()
                    .map(|ary| {
                        let mut vec = Vec::<i64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv as i64);
                                }
                                None => {}
                            }
                        }
                        VddValue::M201(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            //
            if value == VddValue::NULL {
                value = s
                    .u64()
                    .map(|ary| {
                        let mut vec = Vec::<i64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv as i64);
                                }
                                None => {}
                            }
                        }
                        VddValue::M201(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            // 浮点数
            if value == VddValue::NULL {
                value = s
                    .f32()
                    .map(|ary| {
                        let mut vec = Vec::<f64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv as f64);
                                }
                                None => {}
                            }
                        }
                        VddValue::M202(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            //
            if value == VddValue::NULL {
                value = s
                    .f64()
                    .map(|ary| {
                        let mut vec = Vec::<f64>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(vv);
                                }
                                None => {}
                            }
                        }
                        VddValue::M202(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            // 日期
            if value == VddValue::NULL {
                value = s
                    .date()
                    .map(|ary| {
                        let mut vec = Vec::<NaiveDate>::new();
                        let days_offset = NaiveDate::default().num_days_from_ce();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(
                                        NaiveDate::from_num_days_from_ce_opt(vv + days_offset)
                                            .unwrap(),
                                    );
                                }
                                None => {}
                            }
                        }
                        VddValue::M301(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            // 日期时间
            if value == VddValue::NULL {
                value = s
                    .datetime()
                    .map(|ary| {
                        let mut vec = Vec::<NaiveDateTime>::new();
                        for v in ary.into_iter() {
                            match v {
                                Some(vv) => {
                                    vec.push(DateTime::from_timestamp(vv, 0).unwrap().naive_utc());
                                }
                                None => {}
                            }
                        }
                        VddValue::M303(vec)
                    })
                    .unwrap_or(VddValue::NULL);
            }
            value
        }
        _ => {
            tracing::warn!(
                "Polars data type is {}, not supported",
                av.dtype().to_string()
            );
            VddValue::S101(av.to_string())
        }
    }
}

fn fv_to_anyvalue(fv: &VddValue) -> AnyValue {
    match fv {
        VddValue::S101(v)
        | VddValue::S102(v)
        | VddValue::S103(v)
        | VddValue::S801(v)
        | VddValue::EncryptedB64(v) => AnyValue::StringOwned(PlSmallStr::from(v)),
        VddValue::M101(v) | VddValue::M801(v) => {
            let vv = Series::new(PlSmallStr::from_str("0"), v);
            AnyValue::List(vv)
        }
        VddValue::S201(v) | VddValue::S212(v) => AnyValue::from(*v),
        VddValue::M201(v) => {
            let vv = Series::new(PlSmallStr::from_str("0"), v);
            AnyValue::List(vv)
        }
        VddValue::S202(v) => AnyValue::from(*v),
        VddValue::M202(v) => {
            let vv = Series::new(PlSmallStr::from_str("0"), v);
            AnyValue::List(vv)
        }
        VddValue::S211(v) => AnyValue::from(*v),
        VddValue::S291(v) => AnyValue::from(*v),
        VddValue::S301(v) => {
            let days = v.num_days_from_ce() - NaiveDate::default().num_days_from_ce();
            AnyValue::Date(days)
        }
        VddValue::M301(v) => {
            let vv = Series::new(PlSmallStr::from_str("0"), v);
            AnyValue::List(vv)
        }
        VddValue::S302(v) => {
            let time = (v.hour() as i64) * 3600 + (v.minute() as i64 * 60) + v.second() as i64;
            AnyValue::Time(time)
        }
        VddValue::M302(v) => {
            let vv = v
                .iter()
                .map(|x| (x.hour() as i64) * 3600 + (x.minute() as i64 * 60) + x.second() as i64)
                .collect::<Vec<i64>>();
            let vv = Series::new(PlSmallStr::from_str("0"), vv);
            AnyValue::List(vv)
        }
        VddValue::S303(v) => {
            let ts = v.and_local_timezone(Utc).unwrap().timestamp_millis();
            AnyValue::Datetime(ts, TimeUnit::Milliseconds, None)
        }
        VddValue::M303(v) => {
            let vv = Series::new(PlSmallStr::from_str("0"), v);
            AnyValue::List(vv)
        }
        VddValue::S401 { v: _, d } => AnyValue::StringOwned(PlSmallStr::from(d)),
        VddValue::M401 { v: _, d } => {
            let file_names = d.join(";");
            AnyValue::StringOwned(PlSmallStr::from(file_names))
        }
        VddValue::S901(v) | VddValue::S903(v) => AnyValue::from(v.0),
        VddValue::M901(v) | VddValue::M903(v) => {
            let vv = v.iter().map(|item| item.0).collect::<Vec<u64>>();
            let vv = Series::new(PlSmallStr::from_str("0"), vv);
            AnyValue::List(vv)
        }
        VddValue::S911(v) => AnyValue::from(v.0),
        VddValue::M911(v) => {
            let vv = v.iter().map(|item| item.0).collect::<Vec<u64>>();
            let vv = Series::new(PlSmallStr::from_str("0"), vv);
            AnyValue::List(vv)
        }
        VddValue::NULL => AnyValue::Null,
        VddValue::S501 => AnyValue::Null,
    }
}

/// 从dataframe中生成记录矩阵
pub(crate) fn records_from_df(
    model_id: u64,
    user_id: UserId,
    fields: &[VddFieldRef],
    df: &DataFrame,
) -> Result<Vec<VddRecord>, Astr> {
    let mut result: Vec<VddRecord> = Vec::with_capacity(df.height());
    for row_index in 0..df.height() {
        let mut record = VddRecord::new_default(user_id.clone(), model_id);
        df.get_columns().iter().for_each(|x| {
            let current_field = fields
                .iter()
                .find(|field| field.label == x.name().as_str())
                .unwrap();
            let v = fv_from_anyvalue(&x.get(row_index).unwrap());
            record.value.insert(current_field.id.clone(), v.into());
        });
        result.push(record);
    }
    Ok(result)
}
